package com.fwmagic.flink.streaming;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/***
 * Task(任务)划分,及subTask
 * DataFlow Graph中任务划分依据
 * 1、算子的并行度发生了改变（rebalance）
 * 2、执行了keyBy,broadcast等操作，回发生Hash（shuffle）操作
 * 3、调用了startNewChain方法
 * 4、调用了disableChaining方法
 * .disableChaining();将这个算子单独划分处理，生成一个Task，和其他算子不再有Operator Chain
 * .startNewChain();从该方法开始，开启一个新的执行链，从这个算子之前，发生redistriting
 */
public class StreamingDemoSlotSharingGroupTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> dataSource = env.socketTextStream("localhost", 8888);

        SingleOutputStreamOperator<String> words = dataSource.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                out.collect(value);
            }
        }).slotSharingGroup("fwmagic");

        SingleOutputStreamOperator<String> filtered = words.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                return value.startsWith("h");
            }
        });

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = filtered.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return Tuple2.of(value, 1);
            }
        });

        SingleOutputStreamOperator<Tuple2<String, Integer>> sumed = wordAndOne.keyBy(0).sum(1);

        sumed.print();

        String executionPlan = env.getExecutionPlan();
        System.err.println(executionPlan);

        env.execute("StreamingDemoChainTest");


    }
}
